storage: fix fetchConsumer panic when a topic is deleted and recreated concurrently - #865
Open
ZayanKhan-12 wants to merge 1 commit into
Open
Conversation
…d concurrently fetchConsumer snapshots the consumer group's topics under the consumer lock, releases it, and then reads broker offsets under the broker lock. Storage requests are spread across workers (SetDeleteTopic and SetBrokerOffset go to a random worker, SetConsumerOffset and FetchConsumer to a hashed one), so a topic can be deleted and recreated between the two phases. The broker view of the topic then disagrees with the consumer snapshot in two ways that panic the process: - a partition ring can exist with no stored offsets yet (deleteTopic removes consumer topics and broker state under separate locks, so a consumer commit that lands in between survives the delete; when the recreated topic's partition list is grown by addBrokerOffset, rings for partitions without received offsets are empty), making partition.BrokerOffsets empty and the read of BrokerOffsets[len(BrokerOffsets)-1] panic with index out of range [-1] - the recreated topic can have fewer partitions than the consumer snapshot, so indexing topicMap[p] panics with index out of range Guard both: skip partitions the broker no longer knows about, and only compute current lag when at least one broker offset is present. The consumer offsets are still returned either way, consistent with how a fully deleted topic is already handled. Both states are covered by regression tests that reproduce the exact panic prior to this fix. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes #864
Problem
fetchConsumersnapshots the consumer group's topics underconsumerLock, releases it, then reads broker offsets underbrokerLock. Because storage requests are spread across workers (SetDeleteTopic/SetBrokerOffsetgo to a random worker,SetConsumerOffset/FetchConsumerto a hashed one), a topic delete + recreate can interleave between the two phases, leaving the broker view inconsistent with the consumer snapshot:deleteTopicsurvives;addBrokerOffsetgrows the recreated topic's partition list with empty rings).partition.BrokerOffsetsis then empty, and sincegetConsumerTopicListreturnsOffsetsat full ring length (nils included), thelen(partition.Offsets) > 0guard passes andBrokerOffsets[len(BrokerOffsets)-1]panics withindex out of range [-1](inmemory.go:871).topicMap[p](inmemory.go:863) indexes out of range.Either way the whole process dies. Full analysis in #864.
Fix
Mirror the existing "topic may have just been deleted" handling a few lines above:
p >= len(topicMap))CurrentLagwhen at least one broker offset is presentConsumer offsets are still returned in both cases.
Testing
TestInMemoryStorage_fetchConsumer_Expired). Both reproduce the exact panic before the fix:go test -race ./...: all packages pass.go vet ./...andgofmtclean.🤖 Generated with Claude Code